package rxgo

import (
	"context"
	"reflect"
	"fmt"
	"time"
	"sync"
)

type filteringOperator struct {
	opFunc func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool)
}

func (ftop filteringOperator) op(ctx context.Context, o *Observable) {
	// must hold defintion of flow resourcs here, such as chan etc., that is allocated when connected
	// this resurces may be changed when operation routine is running.
	in := o.pred.outflow
	out := o.outflow
	//fmt.Println(o.name, "operator in/out chan ", in, out)
	var wg sync.WaitGroup

	go func() {
		end := false
		for x := range in {
			if end {
				break
			}
			// can not pass a interface as parameter (pointer) to gorountion for it may change its value outside!
			xv := reflect.ValueOf(x)
			// send an error to stream if the flip not accept error
			if e, ok := x.(error); ok && !o.flip_accept_error {
				o.sendToFlow(ctx, e, out)
				continue
			}
			// scheduler
			switch threading := o.threading; threading {
			case ThreadingDefault:
				if ftop.opFunc(ctx, o, xv, out) {
					end = true
				}
			case ThreadingIO:
				fallthrough
			case ThreadingComputing:
				wg.Add(1)
				go func() {
					defer wg.Done()
					if ftop.opFunc(ctx, o, xv, out) {
						end = true
					}
				}()
			default:
			}
		}
		
		if o.flip != nil {
			buf := reflect.ValueOf(o.flip)
			if buf.Kind() != reflect.Slice {
				panic("observable flip usage error")
			}
			for i := 0; i < buf.Len(); i++ {
				o.sendToFlow(ctx, buf.Index(i).Interface(), out)
			}
		}

		wg.Wait() //waiting all go-routines completed
		o.closeFlow(out)
	}()
}

func (parent *Observable) Debounce(timespan time.Duration) (o *Observable) {
	o = parent.newFilterObservable("debounce")
	o.operator = debounceOperator
	debounce_timespan = timespan
	return o
}

var debounce_count = 0
var debounce_timespan time.Duration
var debounceOperator = filteringOperator{func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
	debounce_count++
	go func() {
		temp := debounce_count
		time.Sleep(debounce_timespan)
		select {
		case <-ctx.Done():
			return
		default:
			if temp == debounce_count {
				o.sendToFlow(ctx, item.Interface(), out)
			}
		}
	}()
	return false
}}

func (parent *Observable) Distinct() (o *Observable) {
	o = parent.newFilterObservable("distinct")
	o.operator = distinctOperator
	return o
}

var distinct_map = map[string]bool{}
var distinctOperator = filteringOperator{func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool){
	key := fmt.Sprintf("%v", item)
	_, ok := distinct_map[key]
	if !ok {
		distinct_map[key] = true
		o.sendToFlow(ctx, item.Interface(), out)
	}
	return false
}}

func (parent *Observable) ElementAt(num int) (o *Observable) {
	o = parent.newFilterObservable("elementAt")
	o.operator = elementAtOperator
	elemat_n = num
	return o
}

var elemat_count = 0
var elemat_n = 0
var elementAtOperator = filteringOperator{func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
	if elemat_count == elemat_n {
		o.sendToFlow(ctx, item.Interface(), out)
		return true
	}
	elemat_count++
	return false
}}

func (parent *Observable) Filter(predicate func(interface{}) bool) (o *Observable) {
	o = parent.newFilterObservable("filter")
	o.operator = filterOperator
	filter_predicate = predicate
	return o
}

var filter_predicate func(interface{}) bool
var filterOperator = filteringOperator{func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
	if filter_predicate(item.Interface()) {
		o.sendToFlow(ctx, item.Interface(), out)
	}
	return false
}}

func (parent *Observable) First() (o *Observable) {
	o = parent.newFilterObservable("first")
	o.operator = firstOperator
	return o
}

var firstOperator = filteringOperator{func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
	o.sendToFlow(ctx, item.Interface(), out)
	return true
}}

func (parent *Observable) IgnoreElements() (o *Observable) {
	o = parent.newFilterObservable("ignoreElements")
	o.operator = ignoreElementsOperator
	return o
}

var ignoreElementsOperator = filteringOperator{func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
	return false
}}

func (parent *Observable) Last() (o *Observable) {
	o = parent.newFilterObservable("last")
	o.operator = lastOperator
	return o
}

var lastOperator = filteringOperator{func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
	o.flip = append([]interface{}{}, item.Interface())
	return false
}}

func (parent *Observable) Skip(num int) (o *Observable) {
	o = parent.newFilterObservable("skip")
	o.operator = skipOperator
	skip_n = num
	return o
}

var skip_count = 0
var skip_n = 0
var skipOperator = filteringOperator{func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
	skip_count++
	if skip_count > skip_n {
		o.sendToFlow(ctx, item.Interface(), out)
	}
	return false
}}

func (parent *Observable) SkipLast(num int) (o *Observable) {
	o = parent.newFilterObservable("skipLast")
	o.operator = skipLastOperator
	skipLast_n = num
	return o
}

var skipLast_count = 0
var skipLast_n = 0
var skipLast_buf []interface{}
var skipLastOperator = filteringOperator{func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
	skipLast_count++
	skipLast_buf = append(skipLast_buf, item.Interface())
	if skipLast_count > skipLast_n {
		o.sendToFlow(ctx, skipLast_buf[0], out)
		skipLast_buf = skipLast_buf[1:]
	}
	return false
}}

func (parent *Observable) Take(num int) (o *Observable) {
	o = parent.newFilterObservable("take")
	o.operator = takeOperator
	take_n = num
	return o
}

var take_count = 0
var take_n = 0
var takeOperator = filteringOperator{func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
	take_count++
	if take_count > take_n {
		return true
	}
	o.sendToFlow(ctx, item.Interface(), out)
	return false
}}

func (parent *Observable) TakeLast(num int) (o *Observable) {
	o = parent.newFilterObservable("takeLast")
	o.operator = takeLastOperator
	takeLast_n = num
	return o
}

var takeLast_count = 0
var takeLast_n = 0
var takeLast_buf []interface{}
var takeLastOperator = filteringOperator{func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool) {
	takeLast_count++
	if takeLast_count > takeLast_n {
		takeLast_buf = takeLast_buf[1:]
	}
	takeLast_buf = append(takeLast_buf, item.Interface())
	o.flip = takeLast_buf
	return false
}}

func (parent *Observable) newFilterObservable(name string) (o *Observable) {
	//new Observable
	o = newObservable()
	o.Name = name

	//chain Observables
	parent.next = o
	o.pred = parent
	o.root = parent.root

	//set options
	o.buf_len = BufferLen
	o.flip = nil
	o.flip_accept_error = true
	o.flip_sup_ctx = true
	return o
}